package org.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.AtguiguUtil;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;

import java.util.HashMap;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.log
 * Author: ZengHaiFeng
 * CreateTime:2023/4/22 16:42:09
 * Description: Todo
 * Version：1.0
 */
@Slf4j
public class DwdBaseLog extends BaseApp {
    private static final String START = "start";
    private static final String DISPLAY = "displays";
    private static final String ACTION = "actions";
    private static final String ERR = "err";
    private static final String PAGE = "page";

    public static void main(String[] args) {
        new DwdBaseLog().init(20002,2, Constant.TOPIC_ODS_LOG,"DwdBaseLog");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        SingleOutputStreamOperator<String> etlFinalData = etlData(ds);
        SingleOutputStreamOperator<JSONObject> validatedStream = validateNewOrOld(etlFinalData);
        HashMap<String, DataStream<JSONObject>> splitFinalStream = splitStream(validatedStream);
        writeToKafka(splitFinalStream);
    }

    /**
     * @description: write data to Kafka topic .
     * @author: zenghaifeng
     * @date: 2023/4/23 16:27
     * @param splitFinalStream
     **/
    private void writeToKafka(HashMap<String, DataStream<JSONObject>> splitFinalStream) {
        splitFinalStream.get(START).map(JSONAware::toJSONString).sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        splitFinalStream.get(ACTION).map(JSONAware::toJSONString).sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        splitFinalStream.get(DISPLAY).map(JSONAware::toJSONString).sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        splitFinalStream.get(ERR).map(JSONAware::toJSONString).sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        splitFinalStream.get(PAGE).map(JSONAware::toJSONString).sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
    }

    /**
     * @param validatedStream
     * @return SingleOutputStreamOperator<JSONObject>
     * @description: split streams : one stream -> five streams [start<main>,page,display,action,err]
     * @author: zenghaifeng
     * @date: 2023/4/22 20:00
     **/
    private HashMap<String, DataStream<JSONObject>> splitStream(
            SingleOutputStreamOperator<JSONObject> validatedStream) {
        OutputTag<JSONObject> displayOutputTag = new OutputTag<JSONObject>(DISPLAY) {};
        OutputTag<JSONObject> actionOutputTag = new OutputTag<JSONObject>(ACTION) {};
        OutputTag<JSONObject> errOutputTag = new OutputTag<JSONObject>(ERR) {};
        OutputTag<JSONObject> pageOutputTag = new OutputTag<JSONObject>(PAGE) {};
        SingleOutputStreamOperator<JSONObject> startStream = validatedStream.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(
                    JSONObject value,
                    ProcessFunction<JSONObject, JSONObject>.Context ctx,
                    Collector<JSONObject> out) {

                JSONObject common = value.getJSONObject("common");
                Long ts = value.getLong("ts");
                JSONObject start = value.getJSONObject(START);
                if (start != null) {
                    out.collect(value);
                }

                // process page log
                JSONObject page = value.getJSONObject(PAGE);
                if (page != null) {
                    ctx.output(pageOutputTag, value);
                }

                // process display logs
                JSONArray display = value.getJSONArray(DISPLAY);
                if (display != null) {
                    for (int i = 0; i < display.size(); i++) {
                        JSONObject dis = display.getJSONObject(i);
                        dis.putAll(common);
                        dis.put("ts", ts);
                        ctx.output(displayOutputTag, dis);
                    }
                    value.remove(DISPLAY);
                }

                // process action logs
                JSONArray action = value.getJSONArray(ACTION);
                if (action != null) {
                    for (int i = 0; i < action.size(); i++) {
                        JSONObject ac = action.getJSONObject(i);
                        ac.putAll(common);
                        ctx.output(actionOutputTag, ac);
                    }
                    value.remove(ACTION);
                }
                // process err logs
                JSONObject err = value.getJSONObject(ERR);
                if (err != null) {
                    ctx.output(errOutputTag, value);
                    value.remove(ERR);
                }
            }
        });
        SideOutputDataStream<JSONObject> acOut = startStream.getSideOutput(actionOutputTag);
        SideOutputDataStream<JSONObject> disOut = startStream.getSideOutput(displayOutputTag);
        SideOutputDataStream<JSONObject> errOut = startStream.getSideOutput(errOutputTag);
        SideOutputDataStream<JSONObject> pageOut = startStream.getSideOutput(pageOutputTag);

        HashMap<String, DataStream<JSONObject>> streamMap = new HashMap<>();

        streamMap.put(ACTION,acOut);
        streamMap.put(START,startStream);
        streamMap.put(DISPLAY,disOut);
        streamMap.put(ERR,errOut);
        streamMap.put(PAGE,pageOut);
        return streamMap;
    }

    /**
     * @description: validate new or old
     * @author: zenghaifeng
     * @date: 2023/4/22 19:31
     * @param etlFinalData
     * @return SingleOutputStreamOperator<JSONObject>
     **/
    private SingleOutputStreamOperator<JSONObject> validateNewOrOld(
            SingleOutputStreamOperator<String> etlFinalData) {
        return etlFinalData.map(JSON::parseObject)
                .keyBy(data -> data.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String> firstVisitDateState;
                    @Override
                    public void open(Configuration parameters) {
                        // state
                        firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<>("firstVisitDate",String.class));
                    }
                    @Override
                    public void processElement(
                            JSONObject value,
                            KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx,
                            Collector<JSONObject> out) throws Exception {

                        JSONObject common = value.getJSONObject("common");
                        String isNew = common.getString("is_new");
                        Long ts = value.getLong("ts");
                        // current record time
                        String today = AtguiguUtil.tsToDate(ts);
                        // get state
                        String state = firstVisitDateState.value();
                        // condition
                        if (isNew.equals("1")){
                            if (state == null){
                                // first visit update state
                                firstVisitDateState.update(today);
                            }else {
                                if (!today.equals(state)){
                                    // record is new account , but is old
                                    // change 1 => 0
                                    common.put("is_new","0");
                                }
                            }
                        }else {
                            if (state == null){
                                firstVisitDateState.update(AtguiguUtil.tsToDate(ts-(60 * 60 * 1000 * 24)));
                            }
                        }
                        out.collect(value);
                    }
                });
    }

    /**
     * @description: etl data stream
     * @author: zenghaifeng
     * @date: 2023/4/22 18:03
     * @param ds
     * @return SingleOutputStreamOperator<String>
     **/
    private SingleOutputStreamOperator<String> etlData(DataStreamSource<String> ds){
        return ds.filter(value -> {
            try {
                JSONObject obj = JSON.parseObject(value);
                return obj.getJSONObject("common") != null;
            }catch (Exception e){
                log.warn(e + " => This stream have some data is not json ");
                return false;
            }
        });
    }
}
